/*
 * Reducer.h
 *
 *  Created on: 2011-7-3
 *      Author: yong
 */

#ifndef REDUCER_H_
#define REDUCER_H_
#include "Config.h"
#include "Typedef.h"
#include <boost/lambda/lambda.hpp>
using boost::lambda::_1;

namespace distrim
{
template<typename K, typename V>
class Reducer
{

public:
	Reducer(const Config &config) :
		m_config(config)
	{
	}
	virtual ~Reducer()
	{
	}
	// Thread entry.
	virtual void operator()()
	{
		while (true)
		{
			reduce();
			m_vecIntervals.push_back(m_timer.elapsed());
			std::cout << "Time: " << m_vecIntervals[m_vecIntervals.size() - 1]
					<< std::endl;
			// Get new parameters.
			if (terminate()) // If converged
			{
				update();
				m_vecIntervals.push_back(m_timer.elapsed());
				std::cout << "Time: " << m_vecIntervals[m_vecIntervals.size()
						- 1] << std::endl;
				double mean = std::accumulate(m_vecIntervals.begin() + 1,
						m_vecIntervals.end(), double())
						/ (m_vecIntervals.size() - 1);
				std::cout << "Averaged time: " << mean << std::endl;
				for (size_t i = 1; i < m_vecIntervals.size(); ++i)
				{
					m_vecIntervals[i] = (m_vecIntervals[i] - mean)
							* (m_vecIntervals[i] - mean);
				}
				std::cout << "Variance: " << std::accumulate(
						m_vecIntervals.begin() + 1, m_vecIntervals.end(),
						double()) / (double) (m_vecIntervals.size() - 1)
						<< std::endl;
				break; // Exit.
			}
			update();
			m_timer.restart();
		}
	}

	const Config &GetConfig() const
	{
		return m_config;
	}

private:

	virtual bool terminate() = 0;
	virtual void reduce() = 0;
	virtual void update() = 0;

private:
	Config m_config;
	mpi::timer m_timer;
	std::vector<double> m_vecIntervals;
};

}

#endif /* REDUCER_H_ */
